#ifndef SAFE_CHANNEL
#define SAFE_CHANNEL

#pragma once

namespace GSE
{
	template <typename _Tp, typename _queueTp = std::deque<_Tp> >
	class safe_queue : private _queueTp
	{
	private:
		typedef boost::mutex mutex;
		typedef boost::condition condition;
		typedef boost::mutex::scoped_lock lock;
	private:
		size_t _max_size;
		mutex _monitor;
		condition _buffer_not_full, _buffer_not_empty;
		volatile bool _may_stop;
	public:
		explicit safe_queue(size_t limit = (size_t)-1) : _max_size(limit < 1 ? 1 : limit), _may_stop(false) {}

		safe_queue(_queueTp& queue) 
			: _queueTp(queue), _max_size(queue.max_size()) {}

		virtual ~safe_queue(){}

		// for consumer thread...
		_Tp poll()
		{
			lock lk(_monitor);
			while (!_may_stop && 0 == ((_queueTp *)this)->size())
			{
				_buffer_not_empty.wait(lk);
			}
			// check if caller intentionally calls for stop
			if(_may_stop && 0 == ((_queueTp *)this)->size())
			{
				throw std::logic_error("consume to end");
			}
			// pop back
			_Tp item = pop();
			_buffer_not_full.notify_one();
			return item;
		}
		// for producer thread...
		bool offer(_Tp item)
		{
			lock lk(_monitor);
			while (_max_size == ((_queueTp *)this)->size())
			{
				_buffer_not_full.wait(lk);
			}
			// push front
			push(item);
			_buffer_not_empty.notify_one();
			return true;
		}
		virtual void may_stop(bool bMayStop = true)
		{
			lock lk(_monitor);
			_may_stop = bMayStop;
			if(bMayStop)
			{
				// if producers exit, wake up the
				// waiting consumer threads...
				_buffer_not_empty.notify_all();
			}
		}

		// for outside callers only, calling this on Linux 7.2 within 
		// the above poll() or offer() will deadlock 
		typename _queueTp::size_type size() 
		{
			lock lk(_monitor);
			return ((_queueTp *)this)->size();
		}

	protected:
		virtual _Tp pop() // retrieve from end
		{
			_Tp item = ((_queueTp *)this)->back();
			((_queueTp *)this)->pop_back(); 
			return item;
		}

		virtual void push(const _Tp item) // insert at head...
		{
			((_queueTp *)this)->insert(_queueTp::begin(), item);
		}
	};
}

#endif  // endif of SAFE_CHANNEL

